package com.jmemoir.mq.rabbitmq.utils;

import lombok.AllArgsConstructor;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * RabbitMQ 工具类
 *
 * @author Tellsea
 * @date 2023/9/5
 */
@Component
@AllArgsConstructor
public class RabbitMqUtils {

    private final RabbitTemplate rabbitTemplate;

    /**
     * 简单模型
     * <p>
     * 简单模型-simple：生产者直接发送到队列
     * 简单模型-work：生产者直接发送到队列，有多个消费者
     *
     * @param routingKey 路由Key
     * @param message    消息
     */
    public void sendMessage(String routingKey, Object message) {
        rabbitTemplate.convertAndSend(routingKey, message, getCorrelationData());
    }

    /**
     * 订阅模型
     * <p>
     * 订阅模型-Fanout：生产者 > 交换机 > 发送给所有绑定交换机的队列
     * rabbitTemplate.convertAndSend(exchange, "", message);
     * 订阅模型-Direct：生产者 > 交换机 > 发送给所有绑定交换机、且路由Key相等的队列
     * 订阅模型-Topic：生产者 > 交换机 > 发送给所有绑定交换机、且路由Key相等的队列（支持通配符）
     *
     * @param exchange   交换机
     * @param routingKey 路由Key
     * @param message    消息
     */
    public void sendMessage(String exchange, String routingKey, Object message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message, getCorrelationData());
    }

    /**
     * 延迟队列
     *
     * @param exchange
     * @param routingKey
     * @param message
     * @param delayTime
     */
    public void sendDelayMessage(String exchange, String routingKey, Object message, Integer delayTime) {
        rabbitTemplate.convertAndSend(
                exchange,
                routingKey,
                message,
                processor -> {
                    // 设置消息持久化
                    processor.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                    // 配置消息的过期时间
                    processor.getMessageProperties().setHeader("x-delay", delayTime);
                    // 配置编码格式
                    processor.getMessageProperties().setContentEncoding("utf-8");
                    return processor;
                },
                getCorrelationData()
        );
    }

    /**
     * 每条消息独立的ID
     *
     * @return
     */
    public CorrelationData getCorrelationData() {
        return new CorrelationData(UUID.randomUUID().toString());
    }

    /**
     * 消息中添加消息ID
     *
     * @param json
     * @return
     */
    public Message getMessage(String json) {
        return MessageBuilder
                .withBody(json.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                .setContentEncoding("UTF-8")
                .setMessageId(UUID.randomUUID().toString())
                .build();
    }
}